//package com.dataextend.config;
//
//import com.fasterxml.jackson.databind.deser.std.StringDeserializer;
//import org.apache.kafka.clients.consumer.ConsumerConfig;
//import org.apache.kafka.clients.consumer.KafkaConsumer;
//import org.apache.kafka.common.serialization.IntegerDeserializer;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
//import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
//import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
//import org.springframework.boot.context.properties.EnableConfigurationProperties;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//import org.springframework.kafka.annotation.EnableKafka;
//import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
//import org.springframework.kafka.config.KafkaListenerContainerFactory;
//import org.springframework.kafka.core.ConsumerFactory;
//import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
//import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
//import org.springframework.kafka.listener.ContainerProperties;
//
//import java.util.HashMap;
//import java.util.Map;
//
//@Configuration
//@EnableKafka
//@EnableConfigurationProperties(KafkaProperties.class)
//public class KafkaConfig {
//    @Value("${kafka.bootstrap.servers}")
//    private String kafkaBootstrapServers;
//    @Value("${session.timeout.ms}")
//    private Integer sessionTimeoutMs;
//    @Value("${enable.auto.commit}")
//    private boolean enableAutoCommit;
//    @Value("${auto.commit.interval.ms}")
//    private Integer autoCommitIntervalMs;
//    @Value("${auto.offset.reset}")
//    private String autoOffsetReset;
//    @Value("${group.id}")
//    private String groupId;
//
////    @Bean
////    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactory() {
////        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
////        factory.setConsumerFactory(consumerFactory());
////        factory.setConcurrency(3);
////        factory.getContainerProperties().setPollTimeout(3000);
////        return factory;
////    }
////
////    @Bean
////    public ConsumerFactory<Integer, String> consumerFactory() {
////        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
////    }
////
////    @Bean
////    public Map<String, Object> consumerConfigs() {
////        Map<String, Object> props = new HashMap<>();
////        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServers);
////        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
////        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
////        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitIntervalMs);
////        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeoutMs);
////        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
////        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
////        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
////        return props;
////    }
////
////    @Bean
////    public KafkaConsumer consumer() {
////        return new KafkaConsumer();
////    }
//
//    @Bean(name = "arsenalKafkaContainerFactory")
//    @ConditionalOnClass(value = ConcurrentKafkaListenerContainerFactoryConfigurer.class)
//    public ConcurrentKafkaListenerContainerFactory<?,?>
//    kafkaListenerContainerFactory(ConcurrentKafkaListenerContainerFactoryConfigurer configurer){
//        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
//                new ConcurrentKafkaListenerContainerFactory<>();
//        configurer.configure(factory, consumerFactory());
//        return factory;
//    }
//
//
//    private Map<String, Object> consumerProps() {
//        Map<String, Object> map = new HashMap<>();
//        map.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "121.4.176.191:9092");
//        map.put(ConsumerConfig.GROUP_ID_CONFIG, "userprofile");
//        map.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//        map.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//        map.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
//        return map;
//    }
//
//
//    public ConsumerFactory<Object, Object> consumerFactory() {
//        return new DefaultKafkaConsumerFactory<>(consumerProps());
//    }
//
//
//    @Bean(name = "ManualKafkaContainerFactory")
//    public ConcurrentKafkaListenerContainerFactory<?,?> manualKafkaContainerFactory() {
//        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
//                new ConcurrentKafkaListenerContainerFactory<>();
//        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
//        factory.setConsumerFactory(manualFactory());
//        return factory;
//    }
//
//
//    public ConsumerFactory<Object, Object> manualFactory() {
//        Map<String, Object> map = consumerProps();
//        map.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
//        return new DefaultKafkaConsumerFactory<>(map);
//    }
//
//
//
//
//}
